package com.kyp.basic;

import java.io.IOException;

import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;
import org.apache.log4j.Logger;

public class BasicCalcReducer extends
		Reducer<BasicCalcKey, DoubleWritable, Text, DoubleWritable> {

	private static final Logger sLogger = Logger.getLogger(BasicCalcReducer.class);
			
	private MultipleOutputs<Text, DoubleWritable> mMos;
	
	private Text mOutputKey = new Text();

	private DoubleWritable mResult = new DoubleWritable();

	
	@Override
	protected void setup(Context context) throws IOException,
			InterruptedException {
		
		sLogger.info("***************Reducer Start***************");
		mMos = new MultipleOutputs<Text, DoubleWritable>(context);
	}

	@Override
	protected void reduce(BasicCalcKey key, Iterable<DoubleWritable> values, Context context){

		
		BasicCalcBean calBean = new BasicCalcProcess.Builder()
													.value(values)
													.build()
													.execute();
		
		
		writeToFile(key.toString()+"[" + BasicCalcBean.KEY_SUM + "]", calBean.getSum());
		writeToFile(key.toString()+"[" + BasicCalcBean.KEY_AVG + "]", calBean.getAvg());
		writeToFile(key.toString()+"[" + BasicCalcBean.KEY_COUNT + "]", calBean.getCount());
		writeToFile(key.toString()+"[" + BasicCalcBean.KEY_MAX + "]", calBean.getMax());
		writeToFile(key.toString()+"[" + BasicCalcBean.KEY_MIN + "]", calBean.getMin());
		writeToFile(key.toString()+"[" + BasicCalcBean.KEY_COUNT + "]", calBean.getCount());
		writeToFile(key.toString()+"[" + BasicCalcBean.KEY_GEOMEAN + "]", calBean.getGeoMean());
		writeToFile(key.toString()+"[" + BasicCalcBean.KEY_MEDIAN + "]", calBean.getMed());
		writeToFile(key.toString()+"[" + BasicCalcBean.KEY_STDDEV + "]", calBean.getStddev());
		writeToFile(key.toString()+"[" + BasicCalcBean.KEY_VARIANCE + "]", calBean.getVar());
		
	}
	
	private void writeToFile(String key, Double value){
		mOutputKey.set(key.getBytes());
		mResult.set(value);
		try {
			mMos.write(BasicCalc.RESULT_FILE_NAME, mOutputKey, mResult);
		} catch (IOException e) {
			sLogger.error(e.getMessage(), e);
		} catch (InterruptedException e) {
			sLogger.error(e.getMessage(), e);
		}

	}

}
